package index;


import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Comparator;


import xxl.core.collections.containers.Container;
import xxl.core.collections.containers.io.BufferedContainer;
import xxl.core.collections.containers.io.ConverterContainer;
import xxl.core.collections.containers.io.MultiBlockContainer;
import xxl.core.collections.queues.DynamicHeap;
import xxl.core.cursors.Cursor;
import xxl.core.cursors.filters.Taker;
import xxl.core.functions.Function;
import xxl.core.indexStructures.MTree;
import xxl.core.indexStructures.ORTree;
import xxl.core.io.LRUBuffer;
import xxl.core.io.converters.ConvertableConverter;
import xxl.core.io.converters.Converter;
import xxl.core.indexStructures.MTree.Sphere;
import xxl.core.indexStructures.Tree.Query.Candidate;

public class MTreeIndex {

	/**
	 * The precision parameter for the computation of space filling curves.
	 */
	protected static int FILLING_CURVE_PRECISION = 1<<30;
	private static int mtree_minCapacity=1; 
	private static int mtree_maxCapacity=10; 
	private static int bufferSize=10*640; 
	private  MTree mTree;
	private String folderPath; 
	public MTreeIndex(String f){ 
		if (f.endsWith("/")){
			folderPath=f;
		}else {
			folderPath=f+"/";
		}
		mTree=new MTree(MTree.HYPERPLANE_SPLIT); 
		MTree.pointDistance=new SparseMatrixDistance(); 
	}
	/**
	 * An unary factory method that returns a descriptor for
	 * a given point. That means a new Sphere is generated
	 * containing the given point as its center.
	 */
	public static Function getDescriptor = new Function () {
		/**
		 * 
		 */
		private static final long serialVersionUID = 3898138361492264370L;

		public Object invoke (Object o) {
			SparseMatrix point = (SparseMatrix)o; 
			return new Sphere(point, 0.0, centerConverter());
		}
	};
	/**
	 * Returns a converter that serializes the center
	 * of sphere objects. In this case the center of a sphere
	 * is an high-dimensional DoublePoint. <br>
	 * This converter is used by the M-Tree to read/write leaf
	 * nodes to external memory.
	 *
	 * @param dimension the dimension of the DoublePoint representing
	 * 		the center of the sphere
	 * @return a converter serializing DoublePoints
	 */
	public static Converter centerConverter () {
		return new ConvertableConverter(
			new Function () {
				private static final long serialVersionUID = 1L;
				public Object invoke () {
					return new SparseMatrix(); 
				}
			}
		);
	}
	/**
	 * Returns a converter that serializes the descriptors of
	 * the M-Tree, i.e., spheres
	 * This converter is used by the M-Tree to read/write index
	 * nodes to external memory
	 *
	 * @param dimension the dimension of the DoublePoint representing
	 * 		the center of the sphere
	 * @return a converter serializing spheres
	 */
	public static Converter descriptorConverter () {
		return new ConvertableConverter(
			new Function () {
				private static final long serialVersionUID = -3369422652535885335L;
				public Object invoke () {
					return new Sphere(new SparseMatrix(), 0.0, centerConverter());
				}
			}
		);
	}
	/**
	 * Returns a comparator which evaluates the distance of two candidate objects
	 * to the specified <tt>queryObject</tt>. This comparator
	 * is used for nearest neighbor queries and defines an order on candidate-
	 * descriptors. With the help of a priority queue (Min-heap) and this
	 * comparator the nearest neighbor query can be performed.
	 *
	 * @param queryObject a sphere to which the nearest neighbors should be determined
	 * @return a comparator defining an order on candidate objects
	 */
	public static Comparator getDistanceBasedComparator (final Sphere queryObject) {
		return new Comparator () {
			public int compare (Object candidate1, Object candidate2) {
				double sphereDist1 = queryObject.sphereDistance((Sphere)((Candidate)candidate1).descriptor()),
					   sphereDist2 = queryObject.sphereDistance((Sphere)((Candidate)candidate2).descriptor());
				return sphereDist1 < sphereDist2 ? -1 : sphereDist1 == sphereDist2 ? 0 : 1;
			}
		};
	}
	public MTree getMTree(){
		return mTree; 
	}
	public Container reopenIndex() throws IOException{
		DataInputStream dataInput=new DataInputStream(new FileInputStream(folderPath+"/Root.bin"));
		Long rootPageId=new Long(dataInput.readLong());
		int height=dataInput.readInt();
		Sphere rootDescriptor=new Sphere(new SparseMatrix(), 0.0, centerConverter()); 
		rootDescriptor.read(dataInput);
		dataInput.close(); 
		Container container = 
			new BufferedContainer(
				new ConverterContainer(
					// leaf nodes are 16+8 Bytes of size.
					// index nodes are (16+8+8)+8 Bytes of size.
					// ==> so take the maximum for the block size!
					// additionally each node consumes further 4 bytes for
					// node number and 2 bytes for level information.
					new MultiBlockContainer(folderPath+"MTree"),
	
					mTree.nodeConverter(mTree.leafEntryConverter(centerConverter()), mTree.indexEntryConverter(descriptorConverter())) // define node converter
				),	new LRUBuffer(bufferSize),true); 
		ORTree.IndexEntry rootEntry = (ORTree.IndexEntry) ((ORTree.IndexEntry)mTree.createIndexEntry(height)).initialize(rootDescriptor).initialize(rootPageId);
		mTree.initialize(rootEntry, getDescriptor, container, mtree_minCapacity, mtree_maxCapacity);
		return container; 
	}
	public void commit() throws IOException{
		Long rootPageId = (Long)mTree.rootEntry().id();
		Sphere rootDescriptor = (Sphere) mTree.rootDescriptor();
		int height = mTree.height();
		DataOutputStream dataOutput=new DataOutputStream(new FileOutputStream(folderPath+"Root.bin"));
		dataOutput.writeLong(rootPageId.longValue());
		dataOutput.writeInt(height);
		rootDescriptor.write(dataOutput);
		dataOutput.flush(); 
		dataOutput.close(); 
	}
	public  Container openIndex(){
		//int minCapacity=10; 
		//int maxCapacity=50;
		//int bufferSize=100;  		
		Container container =
			new BufferedContainer(
				new ConverterContainer(
					// leaf nodes are 16+8 Bytes of size.
					// index nodes are (16+8+8)+8 Bytes of size.
					// ==> so take the maximum for the block size!
					// additionally each node consumes further 4 bytes for
					// node number and 2 bytes for level information.
					new MultiBlockContainer(folderPath+"MTree", 2+4+64*16*mtree_maxCapacity),
	
					mTree.nodeConverter(mTree.leafEntryConverter(centerConverter()), mTree.indexEntryConverter(descriptorConverter())) // define node converter
				),new LRUBuffer(bufferSize),true);
		mTree.initialize(getDescriptor, container, mtree_minCapacity, mtree_maxCapacity);
		return container; 
	}
	
	public static void main(String[] args){
		try {
			MTreeIndex mtreeIndex=new MTreeIndex("/media/disk/data"); 
			mtreeIndex.testInsert(); 
			mtreeIndex.testOpen();
		}catch (Exception ex){
			ex.printStackTrace(); 
		}
	}
	public void testOpen() throws IOException{
		Container container=reopenIndex();
		SparseMatrix m=new SparseMatrix(10,20); 
		float[] row=new float[]{1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16};
		m.setRow(12, row);
		Sphere queryObject=new Sphere(m, 1.0,null);  

		// consuming the fifty nearest elements concerning the query object at the
		// the target level;
		// the order is determined by the comparator given to the dynamic heap
		// structure realizing a priority queue
		int targetLevel=0; 
		Cursor cursor = new Taker(
				mTree.query(new DynamicHeap(getDistanceBasedComparator(queryObject)), targetLevel),
				50
		);
		int counter = 0;
		while (cursor.hasNext()) {
			Sphere next = (Sphere)((Candidate)cursor.next()).descriptor();
			System.out.println("candidate no. "+(++counter)+": "+next
							 + "\t distance to query object: "+queryObject.centerDistance(next));
		}
		cursor.close();
		container.close(); 

	}
	//static Object rootPageId;  //= mTree.rootEntry().id();
	//static Sphere rootDescriptor; // = (Sphere) mTree.rootDescriptor();
	//static int height; 
	public  void testInsert() throws IOException{
		Container container=openIndex(); 		 
		
		SparseMatrix m=new SparseMatrix(10,20); 
		float[] row=new float[]{1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16}; 
		m.setRow(12, row);
		mTree.insert(m);
		
		row=new float[]{2,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16};
		for (int i=0; i<1; i++){
			m=new SparseMatrix(10,20);
			row[0]=i+10;
			m.setRow(12, row);
			mTree.insert(m); 
		}

		container.flush();
		this.commit(); 
		System.out.println("size of container:"+container.size());

		// Storing important information
		// The following three variables are necessary to restore the mTree.
		// You can store them wherever you want (for example a file).
		
		
		Sphere queryObject=new Sphere(m, 1.0,null);  

		// consuming the fifty nearest elements concerning the query object at the
		// the target level;
		// the order is determined by the comparator given to the dynamic heap
		// structure realizing a priority queue
		int targetLevel=0; 
		Cursor cursor = new Taker(
				mTree.query(new DynamicHeap(getDistanceBasedComparator(queryObject)), targetLevel),
				50
		);
		int counter = 0;
		while (cursor.hasNext()) {
			Sphere next = (Sphere)((Candidate)cursor.next()).descriptor();
		//	System.out.println("candidate no. "+(++counter)+": "+next
		//					 + "\t distance to query object: "+queryObject.centerDistance(next));
		}
		cursor.close();
		
		container.close(); 
	}
}
